/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package org.cassandra;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;
import me.prettyprint.cassandra.service.CassandraClient;
import me.prettyprint.cassandra.service.CassandraClientPool;
import me.prettyprint.cassandra.service.CassandraClientPoolFactory;
import me.prettyprint.cassandra.service.Keyspace;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.ColumnPath;
import org.apache.cassandra.thrift.KeyRange;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.cassandra.thrift.SuperColumn;
/**
 *
 * @author marc
 */
public class CassandraAccess {
    CassandraClientPool pool=null;
    CassandraClient client=null;
    Keyspace keyspace = null;
    public CassandraAccess(String url, int port, String keyspacename){
        try {
            pool = CassandraClientPoolFactory.INSTANCE.get();
            client = pool.borrowClient(url, port);
            keyspace = client.getKeyspace(keyspacename);
        }catch(Exception e){

        }
    }

    public String insertValue(String cf, byte[] sc, String key, byte[] col, String value){
        String message="success";
        try {
            ColumnPath columnPath = new ColumnPath(cf);
            if(sc!=null)
                columnPath.setSuper_column(sc);
            columnPath.setColumn(col);

            // insert
            keyspace.insert(key, columnPath, value.getBytes());

        } catch (Exception ex) {
            //Logger.getLogger(CassandraAccess.class.getName()).log(Level.SEVERE, null, ex);
            message=ex.getMessage();
        }
        return message;
    }

    public Column readValue(String cf, String key, String col){
        try {
            ColumnPath columnPath = new ColumnPath(cf);
            columnPath.setColumn(col.getBytes());

            // read
            Column columnresp = keyspace.getColumn(key, columnPath);
            return columnresp;
        } catch (Exception ex) {
            //Logger.getLogger(CassandraAccess.class.getName()).log(Level.SEVERE, null, ex);

        }
        return null;
    }

    //three objects -> supercolumn, column, value
    public List<SuperColumn> readSuperValues(String cf, String key){
        try {
            ColumnParent columnParent = new ColumnParent(cf);
            //columnPath.setColumn(col.getBytes());
            SlicePredicate slice_predicate=new SlicePredicate();
            SliceRange sliceRange = new SliceRange();
            sliceRange.setStart(new byte[]{});
            sliceRange.setFinish(new byte[]{});
            slice_predicate.setSlice_range(sliceRange);

            // read
            List<SuperColumn> listsupercolumn = keyspace.getSuperSlice(key, columnParent, slice_predicate);

            return listsupercolumn;
        } catch (Exception ex) {
            Logger.getLogger(CassandraAccess.class.getName()).log(Level.SEVERE, null, ex);

        }
        return null;
    }

    public List<Column> sliceValues(String cf, String key){
        byte[] resp = null;
        try {
            ColumnParent clp = new ColumnParent(cf);
            SliceRange sr = new SliceRange(new byte[0], new byte[0], false, 100);
            SlicePredicate sp = new SlicePredicate();
            sp.setSlice_range(sr);
            return keyspace.getSlice(key, clp, sp);
        } catch (Exception ex) {
            //Logger.getLogger(CassandraAccess.class.getName()).log(Level.SEVERE, null, ex);

        }
        return null;
    }

    public void closePoolCassandra(){
        try {
            pool.releaseClient(keyspace.getClient());
        } catch (Exception ex) {
            Logger.getLogger(CassandraAccess.class.getName()).log(Level.SEVERE, null, ex);
        }
    }

    public Map<String,List<Column>> getListByRange(String cf, ArrayList<String> cols) {
        try {
            ColumnParent columnParent=new ColumnParent(cf);

            SlicePredicate slice_predicate=new SlicePredicate();
            List<byte[]> colNames = new ArrayList<byte[]>();
            for(int i=0;i<cols.size();i++){
                String name=cols.get(i);
                colNames.add(name.getBytes());
            }
            slice_predicate.setColumn_names(colNames);

            KeyRange key_range=new KeyRange(100);
            key_range.setStart_key("");
            key_range.setEnd_key("");
            // slices
            Map<String,List<Column>> mapkeys = keyspace.getRangeSlices(columnParent, slice_predicate, key_range);
            return mapkeys;
        } catch (Exception ex) {
            //Logger.getLogger(CassandraAccess.class.getName()).log(Level.SEVERE, null, ex);

        }
        return null;
    }

}
